package rhc.discribute.node.connecter.impl.zookper;

import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.CreateMode;

import rhc.discribute.BaseComponent;
import rhc.discribute.SingleInstanceComponent;
import rhc.discribute.common.ConfigConstants;
import rhc.discribute.node.connecter.impl.ZKPathLockData;
import rhc.discribute.node.connecter.impl.ZookperConnecter;
import rhc.discribute.node.connecter.impl.ZookperExecute;
import rhc.discribute.node.connecter.impl.zookper.watchCallback.MainHostNodeWatchCallback;
import rhc.discribute.node.connecter.impl.zookper.watchCallback.NodeDataWatchCallback;
import rhc.discribute.node.connecter.impl.zookper.watchCallback.NodeJoinWatchCallback;
import rhc.discribute.node.connecter.impl.zookper.watchCallback.ReceiveMessageWatchCallback;
import rhc.discribute.node.connecter.impl.zookper.watchCallback.ReceiveSuccessMessageWatchCallback;
import rhc.discribute.node.connecter.schedule.SendMessageFailCheckSchedule;
import rhc.discribute.node.host.Host;
import rhc.discribute.node.nodeManage.InnerNodeManage;
import rhc.discribute.node.nodeManage.NodeManage.NodeStatus;
import rhc.discribute.node.sendMessage.SendMessageCallback;
import rhc.discribute.node.sendMessage.SendMessageData;

public class ZKExecute extends BaseComponent implements ZookperExecute{

	private ZookperConnecter zkConnecter;
	
	private ZookperOperate zkOperate;
	
	private InnerNodeManage nodeManage;
	
	public ZKExecute(ZookperConnecter zkConnecter,ZookperOperate zkOperate,InnerNodeManage nodeManage){
		this.zkConnecter=zkConnecter;
		this.zkOperate=zkOperate;
		this.nodeManage=nodeManage;
	}
	
	@Override
	public void addHostPersistentChildWatch() throws Exception {
		MyWatcher watcher = new MyWatcher(SingleInstanceComponent.getInstance(NodeJoinWatchCallback.class), this, true);
		zkOperate.watchPathChild(zkConnecter.HOST_PERSISTENT_PATH, watcher);
	}


	/**
	 * 
	 */
	@Override
	public void hostChange() throws Exception {
		List<String> pathList = zkOperate.getChildPathList(zkConnecter.HOST_PERSISTENT_PATH);

		if (null == pathList || pathList.isEmpty()) {
			logger.info("获取到节点为空");
			return;
		}

		Collection<Host> hostSet = nodeManage.getAllHost();
		Set<Host> removeHost = null == hostSet ? new HashSet<Host>() : new HashSet<Host>(hostSet);
		List<Host> addHost = new LinkedList<Host>();
		for (String path : pathList) {
			Host host = zkConnecter.fromHostEphemeralPathGetHost(path);
			if (nodeManage.hostExists(host)) {
				removeHost.remove(host);
			} else {
				addHost.add(host);
			}
		}
		if (!removeHost.isEmpty()) {
			for (Host host : removeHost) {
				nodeManage.removeHost(host);
			}
		}
		if (!addHost.isEmpty()) {
			for (Host host : addHost) {
				String path = zkConnecter.getHostEphemeralPath(host);
				byte[] data = zkOperate.getPathData(path);
				if (null == data) {
					logger.error(host + " 无法加入，获取不到数据");
					continue;
				}
				String pathData = new String(data, "UTF-8");
				List<String> labelList = nodeManage.getLabelList(pathData);
				nodeManage.addHost(host, labelList);
				addHostDataWatch(host);
			}
		}
	}

	@Override
	public void addHostDataWatch(String path) throws Exception {
		MyWatcher watcher = new MyWatcher(SingleInstanceComponent.getInstance(NodeDataWatchCallback.class), this, true);
		byte[] data = zkOperate.watchPathData(path, watcher);
		hostDataChange(path, data);
	}

	@Override
	public void hostDataChange(String path) throws Exception {
		byte[] data = zkOperate.getPathData(path);
		hostDataChange(path, data);
	}

	@Override
	public void addMainNodeWatch() throws Exception {
		MyWatcher watcher = new MyWatcher(SingleInstanceComponent.getInstance(MainHostNodeWatchCallback.class), this,
				true);
		zkOperate.watchPath(zkConnecter.MASTER_HOST_EPHEMERAL_PATH, watcher,false);
	}

	@Override
	public void mainNodeDataChange() throws Exception {
		if (zkOperate.existsPath(zkConnecter.MASTER_HOST_EPHEMERAL_PATH)) {
			String data = new String(zkOperate.getPathData(zkConnecter.MASTER_HOST_EPHEMERAL_PATH), "UTF-8");
			Host host = Host.fromPathGetHost(data);
			nodeManage.setMainHost(host);
			// 如果状态不是为已连接则更新为已连接
			if (nodeManage.getNodeStatus() != NodeStatus.CONNECT) {
				nodeManage.changeNodeStatus(NodeStatus.CONNECT);
			}
		} else if (nodeManage.canMain()) {
			// 如果可以成为主节点则尝试
			createMainNode();
		} else {
			// 如果已经没有可以成为主节点的节点，则更新状态
			Set<Host> canMainHostSet = nodeManage.getLabelHostSet(ConfigConstants.CAN_MAIN);
			if (null == canMainHostSet || canMainHostSet.isEmpty()) {
				nodeManage.changeNodeStatus(NodeStatus.NO_MAIN);
			}
		}
	}

	@Override
	public void mainNodeDisConnect() throws Exception {
		nodeManage.changeNodeStatus(NodeStatus.DIS_CONNECT);
	}

	@Override
	public void addMessageWatch() throws Exception {
		String messagePath = zkConnecter.getMessageEphemeralPath(nodeManage.getCurrentHost());
		MyWatcher watcher = new MyWatcher(SingleInstanceComponent.getInstance(ReceiveMessageWatchCallback.class), this,
				true);
		zkOperate.watchPath(messagePath, watcher,false);
	}

	@Override
	public void receiveMessage() throws Exception {
		Host currentHost=nodeManage.getCurrentHost();
		String messagePath = zkConnecter.getMessageEphemeralPath(currentHost);
		byte[] data = zkOperate.getPathData(messagePath);
		String message = zkConnecter.getStrData(data);

		zkConnecter.touchRecevieMsg(message);
		try {
			// 接收消息后更新节点数据，通知发送节点已接收
			// 先获取到该节点锁再改变数据，发送方会先获取到该节点锁再注册回调，防止注册回调前数据已更改，接收不到回调
			zkOperate.setData(messagePath, data, -1);
		} finally {
		}
	}

	@Override
	public void sendToHostMessageSuccess(String path,ZKPathLockData zkLockData,SendMessageCallback callback) throws Exception {

		byte[] data=null;
		Host toHost=zkConnecter.fromHostEphemeralPathGetHost(path);
		Host host=nodeManage.getCurrentHost();
		boolean success = false;
		//获取到删除消息节点锁
		String lockPath=getHostToHostDeletePathLock(host,toHost);
		if(zkOperate.existsPath(path)){
			ZKPathLockData delZkLockData=zkConnecter.acquire(lockPath);
			try{
				if(zkOperate.existsPath(path)){
					try {
						data=zkOperate.getPathData(path);
						zkOperate.deletNode(path, -1);
						success=true;
					} catch (Exception e) {
						logger.error("对方成功接收消息回调出错");
						e.printStackTrace();
					} finally{
						zkConnecter.release(zkLockData);
					}
				}
			}finally{
				zkConnecter.release(delZkLockData);
			}
		}
		
		String message=zkConnecter.getStrData(data);

		if(success && null != callback){
			zkConnecter.touchSendSuccessMsg(message, toHost, callback);
		}
	}
	
	@Override
	public void sendToHostMessageFail(String path,ZKPathLockData zkLockData,SendMessageCallback callback) throws Exception {
		
		byte[] data=null;
		Host toHost=zkConnecter.fromHostEphemeralPathGetHost(path);
		Host host=nodeManage.getCurrentHost();
		boolean success = false;
		String lockPath=getHostToHostDeletePathLock(host,toHost);
		//如果节点还存在才尝试去释放节点
		if(zkOperate.existsPath(path)){
			ZKPathLockData delZkLockData=zkConnecter.acquire(lockPath);
			try{
				if(zkOperate.existsPath(path)){
					try {
						data=zkOperate.getPathData(path);
						zkOperate.deletNode(path, -1);
						success=true;
					} catch (Exception e) {
						logger.error("对方成功接收消息回调出错");
						e.printStackTrace();
					} finally{
						zkConnecter.release(zkLockData);
					}
				}
			}finally{
				zkConnecter.release(delZkLockData);
			}
		}
		
		String message=zkConnecter.getStrData(data);
		
		if(success && null != callback){
			zkConnecter.touchSendFailMsg(message, toHost, callback);
		}
	}

	@Override
	public void sentMessageToHost(Host host, String message, int tryCount,SendMessageCallback callback,boolean wait) throws Exception {

		if (!nodeManage.hostExists(host)) {
			throw new IllegalArgumentException("节点不存在，" + host);
		}
		// 将数据写入到该节点下
		String messagePath = zkConnecter.getMessageEphemeralPath(host);

		byte[] data = zkConnecter.getByteData(message);
		// 锁路径
		String lockPath = host.toString();
		ZKPathLockData zkLockData=zkConnecter.acquire(lockPath);
		try {
			CountDownLatch latch=new CountDownLatch(1);
			zkOperate.createPath(messagePath, data, false, CreateMode.EPHEMERAL);
			
			SendMessageData callbackData=new SendMessageData(messagePath,latch,this,zkLockData,callback);
			// 如果发送成功则监控数据节点变化，数据改变时表示对方已收到
			MyWatcher watcher = new MyWatcher(
					SingleInstanceComponent.getInstance(ReceiveSuccessMessageWatchCallback.class), callbackData, true);
			zkOperate.watchPathData(messagePath, watcher);
			
			zkConnecter.touchSendMsg(message, host, callback);
			
			if(wait){
				//如果等待则等待3秒
				if(!latch.await(SendMessageFailCheckSchedule.waitSeconds, TimeUnit.SECONDS)){
					callbackData.clear();
					latch.countDown();
					sendToHostMessageFail(messagePath,zkLockData,callback);
				}
			}else{
				//否则注册一个定时任务，3秒后检测是否发送成功
				SendMessageFailCheckSchedule schedule=new SendMessageFailCheckSchedule(callbackData);
				schedule.run();
			}
		} finally {
//			zkConnecter.releasePathLock(lockPath);
		}
	}
	
	/**
	 * 增加一个节点时需要增加对该节点的数据监控
	 * 
	 * @param host
	 * @throws Exception
	 */
	public void addHostDataWatch(Host host) throws Exception {
		String hostPath = zkConnecter.getHostEphemeralPath(host);
		addHostDataWatch(hostPath);
	}

	public void hostDataChange(String path, byte[] newData) throws Exception {
		String strData = new String(newData, "UTF-8");
		logger.debug(path + " 更新数据为 " + strData);

	}
	
	/**
	 * 尝试创建主节点，创建成功则自己成为主节点
	 */
	public void createMainNode() {
		logger.debug("尝试创建主节点");
		try {
			byte[] data = nodeManage.getCurrentHost().toString().getBytes("UTF-8");
			zkOperate.createPath(zkConnecter.MASTER_HOST_EPHEMERAL_PATH, data, true, CreateMode.EPHEMERAL);
			logger.debug("创建主节点成功");
		} catch (Exception e) {
			logger.debug("创建主节点失败");
		}
	}
	
	/**一个节点发送消息到另一个节点需要先获取锁，等另一个节点接收消息触发发送节点的回调后才会释放锁，
	 * 有时候接收节点一直不释放则一直不能再发送消息到该节点，所以做一个定时，当时间到时还没触发则主动删除消息节点，
	 * 释放锁，但是接收消息成功回调也会释放锁，所以做一个锁，只有获得锁的才能删除消息节点，释放锁
	 * @param host
	 * @param toHost
	 * @return
	 */
	private String getHostToHostDeletePathLock(Host host,Host toHost){
		return host+"_"+toHost;
	}

}
